package org.jxkj.app

import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.jxkj.service.PmuSwitchService

/**
 * PMU开关量频率计算
 */
object PmuSwitch {

  /**
   * 计算最大频率、最小频率、是否越界
   * @param ss
   * @param sc
   */
  def calcFreq(ss: SparkSession, sc: SparkContext): Unit = {

    // 加载数据库表
    val jdbcDF1 = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url)
      .option("dbtable", "hms_first_frequency_pmu_point_conf")
      .option("user", MysqlDml.user)
      .option("password", MysqlDml.password)
      .option("driver", MysqlDml.driver)
      .load()
      .cache()
    jdbcDF1.createOrReplaceTempView("hms_first_frequency_pmu_point_conf")

    // 加载数据库表
    val jdbcDF2 = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_cm_high_frequency")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
      .cache()
    jdbcDF2.createOrReplaceTempView("hms_cm_high_frequency")

    // 查询动作测点对应的频率测点
    val df1 = ss.sql(
      s"""
         | select
         |   a.point_code as act_point_code,
         |   b.point_code as freq_point_code
         | from
         |   hms_first_frequency_pmu_point_conf a inner join hms_first_frequency_pmu_point_conf b on
         |     substr(a.point_name, 1, 3) = substr(b.point_name, 1, 3)
         |   and
         |     a.category = 'act'
         |   and
         |     b.category = 'freq'
         |""".stripMargin)
    df1.createOrReplaceTempView("df1")

    // 关联查询
    val df2 = ss.sql(
      s"""
         | select
         |   a.*
         | from
         |   hms_cm_high_frequency a left join hms_first_frequency_pmu_point_conf b on
         |     a.point_code = b.point_code
         | where
         |   b.category = 'act' and a.value = '1'
         | and
         |   (a.max_freq is null or a.min_freq is null or a.over_freq is null)
         | order by
         |   a.point_code, a.change_time
         |""".stripMargin)
    df2.createOrReplaceTempView("df2")

    // 关联查询
    val df3 = ss.sql(
      s"""
         | select
         |   a.frequency_id as frequencyId,
         |   a.point_code as actPointCode,
         |   a.point_name as pointName,
         |   b.freq_point_code as freqPointCode,
         |   cast(a.change_time as bigint) as startTime,
         |   cast(a.end_time as bigint) as endTime
         | from
         |   df2 a left join df1 b on
         |     a.point_code = b.act_point_code
         | order by
         |   a.point_code, a.change_time
         |""".stripMargin)
    df3.createOrReplaceTempView("df3")

    // 计算频率并更新数据
    PmuSwitchService.calcAndSaveFreq(df3)
  }



  case class CalcData1(frequencyId: String, actPointCode: String, pointName: String, freqPointCode: String, startTime: Long, endTime: Long) extends Serializable
  case class CalcData2(frequencyId: String, maxFreq: Double, minFreq: Double, overFreq: String) extends Serializable
}













